package com.lzwk.app.ods;

import com.lzwk.app.function.GetJsonInt;
import com.lzwk.app.function.GetJsonString;
import com.lzwk.base.BaseSql;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @title: OdsOcpxPayOrcer
 * @Author Yeaser
 * @Date: 2022/1/18 14:19
 * @Version 1.0
 * Description: TODO
 */
public class OdsOcpxPayOrder extends BaseSql {
    public static void main(String[] args) {
        new OdsOcpxPayOrder().init(1,"ck","ocpx_new_order");
    }
    @Override
    protected void run(StreamTableEnvironment tenv) {


        tenv.executeSql("create table mysql_pay_order(" +
                " `id` INT, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `fee` INT, " +
                " `product_type` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='10.1.192.67', " +
                "   'port'='3306', " +
                "   'username'='ocpx_galaxy', " +
                "   'password'='FLsdf123', " +
                "   'database-name'='ocpx_elf', " +
                "   'table-name'='pay_order', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none' " +
                ")");

        tenv.createTemporaryFunction("GetJsonString", GetJsonString.class);
        tenv.createTemporaryFunction("GetJsonInt", GetJsonInt.class);

        tenv.executeSql(" CREATE TABLE ocpx_promote_promotesource ( " +
                "  id INT, " +
                "  skin_id INT, " +
                "  hash_id string, " +
                "  product_id INT, " +
                "  wxapp_id INT, " +
                "  name string, " +
                "  meta string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector' = 'jdbc', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver', " +
                "  'url' = 'jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name' = 'promote_promotesource', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        tenv.executeSql("CREATE TABLE ocpx_product_productskin (" +
                "  id INT, " +
                "  name string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='product_productskin', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql("CREATE TABLE ocpx_product_product (" +
                "  id INT, " +
                "  name string, " +
                "  tag_id INT, " +
                "  src_prod_id INT, " +
                "  src_prod_type string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='product_product', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='10', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");

        tenv.executeSql("CREATE TABLE ocpx_wxapp_wxapp (" +
                "  id INT, " +
                "  nickname string, " +
                "  appid string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='wxapp_wxapp', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='20', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql("CREATE TABLE ocpx_public_admin_recordorgref (" +
                "  id int, " +
                "  object_type string, " +
                "  object_id int, " +
                "  org_id int, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='public_admin_recordorgref', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='50', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");


        tenv.executeSql("CREATE TABLE ocpx_tag_info (" +
                "  id INT, " +
                "  tag_name string, " +
                "  primary key (id) not enforced " +
                ") with ( " +
                "  'connector'='jdbc', " +
                "  'url'='jdbc:mysql://10.1.192.67:3306/ocpx_elf?characterEncoding=utf8&serverTimezone=Asia/Shanghai&autoReconnect=true', " +
                "  'table-name'='tag_info', " +
                "  'username'='ocpx_galaxy', " +
                "  'password'='FLsdf123', " +
                "  'lookup.cache.max-rows'='5', " +
                "  'lookup.cache.ttl'='600000' " +
                ")");






        tenv.executeSql("create table ods_ocpx_pay_order_new(" +
                " `payer_id` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `org_id` INT, " +
                " `tag_id` INT, " +
                " `source_name` STRING, " +
                " `skin_name` STRING, " +
                " `product_name` STRING, " +
                " `tag_name` STRING, " +
                " `wxapp_nickname` STRING, " +
                " `src_prod_id` INT, " +
                " `creative_id` STRING, " +
                " `wx_aid` STRING, " +
                " `ks_aid` STRING, " +
                " `open_id` STRING, " +
                " `fee` INT, " +
                " `method` STRING, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `product_type` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `ip` STRING, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `meta` STRING, " +
                " `id` INT, " +
                " `platform_meta` STRING, " +
                " `source_platform` STRING, " +
                "   PRIMARY KEY (`id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='10.2.16.44:8030', " +
                "   'table.identifier'='warehouse.ods_ocpx_pay_order_new', " +
                "   'username'='data_group'," +
                "   'password'='GTCStpJKpELuKFSs'," +
                "   'sink.batch.size'='80000', " +
                "   'sink.batch.interval'='3s' " +
                ")");


        tenv.executeSql(" insert into ods_ocpx_pay_order_new " +
                " select " +
                " t1.`payer_id`, " +
                " t1.`product_id`, " +
                " ifnull(t1.`source_id`, 1) AS `source_id`, " +
                " ifnull(t1.`skin_id`, -1) AS `skin_id`, " +
                " ifnull(t1.`wxapp_id`, -1) AS `wxapp_id`, " +
                " ifnull(t7.`org_id`, -1) AS `org_id`, " +
                " ifnull(t5.`tag_id`, -1) AS `tag_id`, " +
                " ifnull(regexp_replace(t3.`name`,'\\n|\\t',''), '') AS `source_name`, " +
                " ifnull(regexp_replace(t4.`name`,'\\n|\\t',''), '') AS `skin_name`, " +
                " ifnull(regexp_replace(t5.`name`,'\\n|\\t',''), '') AS `product_name`, " +
                " ifnull(regexp_replace(t8.`tag_name`,'\\n|\\t',''), '') AS `tag_name`, " +
                " ifnull(regexp_replace(t6.`nickname`,'\\n|\\t',''), '') AS `wxapp_nickname`, " +
                " ifnull(t5.`src_prod_id`,-1) as src_prod_id, " +
                " ifnull(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'creativeid=(\\\\d+)', 1),'') AS creative_id, " +
                " ifnull(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'wx_aid=(\\\\d+)', 1),'') AS wx_aid, " +
                " ifnull(REGEXP_EXTRACT(GetJsonString(t1.`meta`, 'url'), 'ks_aid=(\\\\d+)', 1),'') AS ks_aid, " +
                " ifnull(GetJsonString(t1.`meta`, 'openid'),'') AS `open_id`, " +
                " if(t1.source='dyshop',GetJsonInt(t1.`meta`,'pay_amount'),t1.`fee`) AS `fee`, " +
                " ifnull(GetJsonString(t1.`meta`, 'method'),'') AS `method`, " +
                " t1.`payee_id`, " +
                " ifnull(regexp_replace(t1.`payee_type`,'\\n|\\t',''),'') AS `payee_type`, " +
                " ifnull(t1.`product_type`,-1), " +
                " t1.`paid`, " +
                " t1.`deliver_status`, " +
                " t1.`refund_status`, " +
                " t1.`create_time`, " +
                " ifnull(t1.`pay_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `pay_time`, " +
                " ifnull(t1.`callback_time`,TO_TIMESTAMP('1970-01-01 08:00:01'))  AS `callback_time`, " +
                " ifnull(regexp_replace(t1.`outer_order_id`,'\\n|\\t',''),'') AS `outer_order_id`, " +
                " ifnull(regexp_replace(t1.`ip`,'\\n|\\t',''),'') AS `ip`, " +
                " ifnull(regexp_replace(t1.`source`,'\\n|\\t',''),'') AS `source`, " +
                " ifnull(t1.`is_notified`,-1) AS `is_notified`, " +
                " ifnull(t1.`update_time`, TO_TIMESTAMP('1970-01-01 08:00:01')) AS `update_time`, " +
                " ifnull(regexp_replace(t1.`openid`,'\\n|\\t',''),'') AS `openid`, " +
                " ifnull(regexp_replace(t1.`meta`,'\\n|\\t',''),'') AS `meta`, " +
                " t1.`id`, " +
                " ifnull(t3.`meta`,'') AS `platform_meta`, " +
                " ifnull(GetJsonString(t3.`meta`,'source_platform'),'') AS `source_platform` " +
                " FROM " +
                "     ( " +
                "     select " +
                "         *, " +
                "         row_number() over (partition by `product_id`,`payer_id` order by paid desc) as rn " +
                "    from mysql_pay_order " +
                "     )t1 " +
                " LEFT JOIN ocpx_promote_promotesource FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t3 " +
                " ON t1.source_id=t3.id " +
                " LEFT JOIN ocpx_product_productskin FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                " ON t1.skin_id=t4.id " +
                " LEFT JOIN ocpx_product_product FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                " ON t1.product_id = t5.id " +
                " LEFT JOIN ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` t6 " +
                " ON t1.wxapp_id = t6.id " +
                " LEFT JOIN ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t7 " +
                " ON t1.product_id=t7.object_id " +
                " AND t7.object_type = 'product' " +
                " LEFT JOIN ocpx_tag_info FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                " ON t5.tag_id=t8.id " +
                " WHERE t5.src_prod_type = 'wk_act' and t1.rn = 1 "
        );
    }
}
